package spring.cloud.tasks.executor.name_space.executor;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import spring.cloud.tasks.common.ExecutorServiceUtil;
import spring.cloud.tasks.common.ThreadFactory;
import spring.cloud.tasks.executor.name_space.NameSpaceConfiguration;
import spring.cloud.tasks.executor.name_space.NameSpaceDataManager;
import spring.cloud.tasks.executor.name_space.NameSpaceDataManagerImpl;
import spring.cloud.tasks.executor.name_space.executor.task.TaskExecutor;
import spring.cloud.tasks.executor.name_space.executor.task.task_item.TaskItemExecuteTimeoutTaskSchedulerExecutor;
import spring.cloud.tasks.executor.name_space.executor.task.task_listener.task_statistics.ExecutorTaskStatisticsReportTaskService;
import spring.cloud.tasks.executor.name_space.executor.task.task_listener.task_statistics.ExecutorTaskStatisticsResetTaskService;
import spring.cloud.tasks.executor.utils.LogUtils;
import spring.cloud.tasks.executor.utils.SystemEnvProperties;

import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;

import static spring.cloud.tasks.common.Threads.sleep;

@Slf4j
@Setter
@Getter
public class Executor {
    static volatile boolean needStopFlag = false;
    /**
     * 一台服务器只允许启动一个
     */
    private static AtomicBoolean inited = new AtomicBoolean(false);
    //
    private String nameSpace;
    private NameSpaceDataManager nameSpaceDataManager;
    private String executorId;
    private ExecutorConnectionReConnectListener executorConnectionReConnectListener;
    private ExecutorInitAndTaskInitService executorInitAndTaskInitService;
    private ExecutorTaskStatisticsResetTaskService executorTaskStatisticsResetTaskService;
    private ExecutorTaskStatisticsReportTaskService executorTaskStatisticsReportTaskService;
    //
    private ReentrantLock lock = new ReentrantLock();
    //
    private Thread restartThread;
    private volatile boolean needRestart = false;
    //
    private volatile boolean isShutdown;

    private Executor(String nameSpace, String executorId) {
        this.nameSpace = nameSpace;
        this.executorId = executorId;
        initRestartThread();
    }

    public static boolean getNeedStopFlag() {
        return needStopFlag;
    }

    /**
     * 一个服务器只能启动一个
     */
    public static Executor buildExecutor(String nameSpace, String executorName) {
        if (!inited.compareAndSet(false, true)) {
            return null;
        }
        return new Executor(nameSpace, executorName);
    }

    private void initRestartThread() {
        final String threadName = executorId + "-restart-thread";
        this.restartThread = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    while (true) {
                        if (isShutdown) {
                            return;
                        }
                        if (needRestart) {
                            try {
                                needRestart = false;
                                //
                                execute();
                            } catch (InterruptedException e) {
                                throw e;
                            } catch (Throwable t) {
                                needRestart = true;
                                LogUtils.error(log, "", "Executor {} reinitialize failed, will retry again", executorId, t);
                            }
                        }
                        sleep();
                    }
                } catch (InterruptedException e) {
                    LogUtils.info(log, "", "{} is interrupted", threadName);
                    Thread.currentThread().interrupt();
                }
            }
        });
        this.restartThread.setName(threadName);
        this.restartThread.setDaemon(false);
        this.restartThread.start();
    }

    public void execute() throws Exception {
        lock.lockInterruptibly();
        try {
            if (isShutdown) {
                return;
            }
            long startTime = System.currentTimeMillis();
            shutdown0();
            needStopFlag = false;//全局可见性。防止底层控制标志不可见。
            try {
                LogUtils.info(log, "", "start to discover from saturn console");
                String connectString = null;//配置中心获取
                if (connectString == null) {
                    LogUtils.error(log, "", "zk connection string is blank!");
                    throw new RuntimeException("zk connection string is blank!");
                }
                // 初始化注册中心
                initnameSpaceDataManager(connectString);
                // 注册Executor
                registerExecutor();
                // 启动零点清0成功数错误数的线程
                LogUtils.info(log, "", "start ResetCountService");
                executorTaskStatisticsResetTaskService = new ExecutorTaskStatisticsResetTaskService(executorId);
                executorTaskStatisticsResetTaskService.start();

                executorTaskStatisticsReportTaskService = new ExecutorTaskStatisticsReportTaskService(executorId);
                executorTaskStatisticsReportTaskService.start();
                //
                // 初始化timeout scheduler
                LogUtils.info(log, "", "start to create timeout scheduler");
                TaskItemExecuteTimeoutTaskSchedulerExecutor.createScheduledThreadPoolExecutor(executorId);
                // 添加新增任务时的回调方法，启动已经存在的任务
                LogUtils.info(log, "", "start to register newJobCallback, and async start existing jobs");
                executorInitAndTaskInitService.registerTaskWatcher();
                //
                LogUtils.info(log, "", "The executor {} start successfully which used {} ms", executorId, System.currentTimeMillis() - startTime);
            } catch (Throwable t) {
                shutdown0();
                throw t;
            }
        } finally {
            lock.unlock();
        }
    }

    private void initnameSpaceDataManager(String connectString) throws Exception {
        try {
            // 初始化注册中心
            LogUtils.info(log, "", "start to init zookeeper");
            NameSpaceConfiguration nameSpaceConfiguration = new NameSpaceConfiguration(connectString, nameSpace, 1000, 3000);
            nameSpaceDataManager = new NameSpaceDataManagerImpl(nameSpaceConfiguration);
            nameSpaceDataManager.init();
            executorConnectionReConnectListener = new ExecutorConnectionReConnectListener(executorId) {
                @Override
                public void onLost() {
                    needRestart = true;
                }
            };
            nameSpaceDataManager.addConnectionStateListener(executorConnectionReConnectListener);
        } catch (Exception e) {
            throw e;
        }
    }

    private void registerExecutor() throws Exception {
        try {
            LogUtils.info(log, "", "start to register executor");
            executorInitAndTaskInitService = new ExecutorInitAndTaskInitService(executorId, nameSpaceDataManager);
            executorInitAndTaskInitService.registerExecutor();
        } catch (Exception e) {
            LogUtils.error(log, "", "register executor error", e);
            throw e;
        }
    }


    /**
     * Executor优雅退出： 把自己从集群中拿掉，现有的任务不停； 一直到全部任务都执行完毕，再真正退出； 设置一定超时时间，如果超过这个时间仍未退出，则强行中止
     */
    public void shutdownGracefully() throws Exception {
        if (isShutdown) {
            return;
        }
        lock.lockInterruptibly();
        try {
            if (isShutdown) {
                return;
            }
            shutdownGracefully0();
            restartThread.interrupt();
            //
            isShutdown = true;
        } finally {
            lock.unlock();
        }
    }

    private void shutdownGracefully0() throws Exception {
        lock.lockInterruptibly();
        needStopFlag = true;//全局可见性。防止底层控制标志不可见。
        try {
            LogUtils.info(log, "", "Try to stop executor {} gracefully", executorId);
            if (executorInitAndTaskInitService != null) {
                executorInitAndTaskInitService.unRegisterTaskWatcher();
            }
            // 先关闭统计信息上报，因为上报的zk结点为servers/xxx/xxx，如果放在下线后再关闭，则导致仍然在上报统计信息，垃圾结点由此而生
            shutdownTaskStatisticsReportTaskService();
            //
            // shutdown timeout-watchdog-thread pool
            TaskItemExecuteTimeoutTaskSchedulerExecutor.shutdownScheduledThreadPoolExecutor(executorId);
            tryCompletedTask();
            shutdownTaskExecutors();
            //可能这期间存在和zk交互的场景,最后关闭没有影响
            if (executorConnectionReConnectListener != null) {
                executorConnectionReConnectListener.close();
            }
            if (nameSpaceDataManager != null) {
                nameSpaceDataManager.removeConnectionStateListener(executorConnectionReConnectListener);
                nameSpaceDataManager.close();
            }
            LogUtils.info(log, "", "executor {} is stopped gracefully", executorId);
        } finally {
            lock.unlock();
        }
    }


    /**
     * Executor关闭
     */
    public void shutdown() throws Exception {
        if (isShutdown) {
            return;
        }
        lock.lockInterruptibly();
        try {
            if (isShutdown) {
                return;
            }
            shutdown0();
            restartThread.interrupt();
            isShutdown = true;
        } finally {
            lock.unlock();
        }
    }

    private void shutdown0() throws Exception {
        lock.lockInterruptibly();
        needStopFlag = true;//全局可见性。防止底层控制标志不可见。
        try {
            LogUtils.info(log, "", "Try to stop executor {}", executorId);
            if (executorInitAndTaskInitService != null) {
                executorInitAndTaskInitService.unRegisterTaskWatcher();
            }
            // 先关闭统计信息上报，因为上报的zk结点为servers/xxx/xxx，如果放在下线后再关闭，则导致ExecutorCleanService执行后，仍然在上报统计信息，垃圾结点由此而生
            shutdownTaskStatisticsReportTaskService();
            TaskItemExecuteTimeoutTaskSchedulerExecutor.shutdownScheduledThreadPoolExecutor(executorId);
            shutdownTaskExecutors();
            //可能这期间存在和zk交互的场景,最后关闭没有影响
            if (executorConnectionReConnectListener != null) {
                executorConnectionReConnectListener.close();
            }
            if (nameSpaceDataManager != null) {
                nameSpaceDataManager.removeConnectionStateListener(executorConnectionReConnectListener);
                nameSpaceDataManager.close();
            }
            LogUtils.info(log, "", "The executor {} is stopped", executorId);
        } finally {
            lock.unlock();
        }
    }

    private void shutdownTaskStatisticsReportTaskService() {

        if (executorTaskStatisticsResetTaskService != null) {
            executorTaskStatisticsResetTaskService.stop();
        }
        if (executorTaskStatisticsReportTaskService != null) {
            executorTaskStatisticsReportTaskService.stop();
        }
    }

    private void tryCompletedTask() {
        Map<String, TaskExecutor> taskIdToTaskExecutorMap = ExecutorIdToTaskIdToTaskExecutorMapMap.getExecutorIdToTaskIdToTaskExecutorMapMap().get(executorId);
        if (taskIdToTaskExecutorMap == null) {
            return;
        }
        Set<Entry<String, TaskExecutor>> entrySet = taskIdToTaskExecutorMap.entrySet();
        if (entrySet == null || entrySet.size() == 0) {
            return;
        }
        long startTime = System.currentTimeMillis();
        do {
            sleep();
            //
            boolean hasRunning = false;
            for (Entry<String, TaskExecutor> entry : entrySet) {
                TaskExecutor taskExecutor = entry.getValue();
                if (taskExecutor != null) {
                    if (taskExecutor.getTaskExecuteEngine().isRunning()) {
                        hasRunning = true;
                        break;
                    }
                }
            }
            if (!hasRunning) {
                break;
            }
        } while (System.currentTimeMillis() - startTime < SystemEnvProperties.SHUTDOWN_TIMEOUT * 1000l);
        long endTime = System.currentTimeMillis();
        LogUtils.info(log, "", "Shutdown phase [blockUntilJobCompletedIfNotTimeout] took {}ms", endTime - startTime);
    }

    private void shutdownTaskExecutors() {
        Map<String, TaskExecutor> taskIdToTaskExecutorMap = ExecutorIdToTaskIdToTaskExecutorMapMap.removeTaskExecutors(executorId);
        if (taskIdToTaskExecutorMap == null) {
            return;
        }
        long startTime = System.currentTimeMillis();
        Iterator<String> iterator = taskIdToTaskExecutorMap.keySet().iterator();
        ExecutorService executorService = Executors.newCachedThreadPool(new ThreadFactory(executorId + "-shutdownTaskExecutors-thread", true));
        List<Future<?>> futureList = new ArrayList<>();
        while (iterator.hasNext()) {
            final String taskId = iterator.next();
            final TaskExecutor taskExecutor = taskIdToTaskExecutorMap.get(taskId);
            if (taskExecutor != null) {
                taskExecutor.needStop();
                futureList.add(executorService.submit(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            taskExecutor.shutdown();
                        } catch (Throwable t) {
                            LogUtils.error(log, taskId, "shutdown JobScheduler error", t);
                        }
                    }
                }));
            }
        }
        for (Future<?> future : futureList) {
            try {
                future.get();
            } catch (Exception e) {
                LogUtils.error(log, "", "wait shutdown job error", e);
            }
        }
        ExecutorServiceUtil.shutdownNow(executorService);
        LogUtils.info(log, "", "Shutdown phase [shutdownJobSchedulers] took {}ms", System.currentTimeMillis() - startTime);
    }


}
